package example;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.net.UnknownHostException;
import java.rmi.RemoteException;
import java.util.ArrayList;

import java.net.*;
import java.rmi.RemoteException;
import java.sql.SQLException;
import java.io.*;
import java.util.*;

import message.Tuple;
import message.UniversalMessage;
import message.MessageInterface.QUERY_TYPE;
import message.UniversalMessage.MSG_TYPE;

public class SplitJoinThread  extends Thread {
	int num_workers = 0;
	int queryId = 0;
	public String[] dataServer;
	public ArrayList<Tuple> worker_results[];
	public ArrayList<Tuple> results;
	private Socket socket = null;
	
	public SplitJoinThread(Socket socket, SimpleProtocol kpp, int queryId) throws RemoteException, Exception {
		
	}
	public String[] getDataServer(String query) {		
		//lookup host and update list of data server accordingly
		String[] dataServer;
		num_workers = 2;		
		//update dataServer
		dataServer = new String[num_workers];
		return dataServer;
	}
	
	public void run() {
		String inputLine ="";
		UniversalMessage msg = new UniversalMessage();
		boolean isFinish = false;
		try {
			ObjectInputStream in = new ObjectInputStream(this.socket.getInputStream());
			ObjectOutputStream out = new ObjectOutputStream(this.socket.getOutputStream());			
			while (!isFinish && (msg = (UniversalMessage)in.readObject()) != null) {
				//from client
				switch(msg.getMsgType()) {
				//client query
				case QUERY:
					processClientQuery(in, msg);
					break;
				//data server send hash join data to server
				//1 chunks or > 1 chunks
				case RESULT:
					if (msg.getNumChunks() ==1)
						processHashJoinData(in, msg);
					else 
						mergeHashJoinData(in, msg);
					break;
				}
				//from data server for hash join query?
			}	
			
		} catch (IOException e) {
			e.printStackTrace();
		} catch (ClassNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}	
	}
	//getting joined data from # chunks
	//delete duplication, merge, return client
	private void mergeHashJoinData(ObjectInputStream in, UniversalMessage msg) {
		// TODO Auto-generated method stub
		
	}
	private void processHashJoinData(ObjectInputStream in, UniversalMessage msg) {
		//data server send hash join result to server
		//get queryid
		//merging all results from all corresponding data servers, delete duplication
		System.out.println("result size "+ msg.getResult().size());
		//server send back to client
	}
	private void processClientQuery(ObjectInputStream clientIn, UniversalMessage msg) {
		//get list of data servers
		msg.setNumChunks(num_workers);
		dataServer = getDataServer(msg.getQueryMsg().getQuery());	
		//balance query
		if(num_workers > 1) {
			RequestThread[] process;		
			process = new RequestThread[num_workers];
//			worker_results = new ArrayList[num_workers];
			//create threads 
			for(int i =0;i<num_workers;i++) {
				process[i]  = new RequestThread(i, msg);//(acptSocket, kpp, ks.tableName, ks.lockTable);
			}
			//- each threads will:
			//- connect to corresponding server
			//- send corresponding query
			//- return results
			for(int i =0;i<num_workers;i++) {
				try {
					process[i].join();
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
			//threadJoin: join these slave threads to get results
			//results store in an array list and return to client by using socket
			//at this point combine result store result
			//naive: linear barrier, merge all results
//			for(int i =0;i<num_workers;i++) {
//				for(int j=0;j<worker_results[i].size();j++)
//					results.add(worker_results[i].get(j));
//			}
		}
		else {
			//just send query to one data server only
			UniversalMessage query = new UniversalMessage();
			query.setMsgType(MSG_TYPE.QUERY);
			query.setQueryType(msg.getQueryType());
			query.setQuery(msg.getQuery());			
			//JOIN query
			if (msg.getQueryMsg().getQueryType() == QUERY_TYPE.JOIN) {
				ArrayList<String> pList = getIdleServers();
				query.setPeerList(pList);
			}
			//data server host, port
			String host = dataServer[0];
			int port = 8888;
			UniversalMessage fromDataServer;
			//send query to data servers
			try {
	            socket = new Socket(host, port);//open socket to server           
	            ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream());
	            ObjectInputStream in = new ObjectInputStream(socket.getInputStream());
//	            BufferedReader stdIn = new BufferedReader(new InputStreamReader(System.in));
                out.writeObject(query);
                String content = "";
			    if ((fromDataServer = (UniversalMessage) in.readObject()) != null)  {
			    	//get result from data server
			    	System.out.println("#result of range query: size = "+ fromDataServer.getResult().size());
//			    	for(int i=0;i<fromDataServer.getResult().size();i++) {
//			    	}
			    	//return result to client
                }
	    	    out.close();
	    	    in.close();
//	    	    stdIn.close();
	    	    socket.close();
	        } catch (UnknownHostException e) {
	            System.err.println("Don't know about host: "+host);
	            System.exit(1);
	        } catch (IOException e) {
	            System.err.println("Couldn't get I/O for the connection to: " + host);
	            System.exit(1);
	        } catch (ClassNotFoundException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
						
			
		}
				
	}
	private ArrayList<String> getIdleServers() {
		return null;
	}
	//this is request thread
	//each thread working with one corresponding worker
	public class RequestThread extends Thread {
		public int thread_id;
		public int query_id;
		public String query = "";
		public Socket socket = null;
		public String host = "";
		public int port = 8888;
		public UniversalMessage msg;
		public RequestThread(int id, UniversalMessage message) {
			thread_id = id;
			msg = message;
		}
		public void run() {
			//connect to worker server
			//request data
			UniversalMessage query = new UniversalMessage();
			query.setMsgType(MSG_TYPE.QUERY);
			query.setQueryType(msg.getQueryType());
			query.setQuery(msg.getQuery());		
			query.setNumChunks(msg.getNumChunks());
			
			//data server host, port
			String host = dataServer[thread_id];
			int port = 8888;
			
			UniversalMessage fromDataServer;
			//send query to data servers
			try {
	            socket = new Socket(host, port);//open socket to data server           
	            ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream());
	            ObjectInputStream in = new ObjectInputStream(socket.getInputStream());
                out.writeObject(query);
			    if ((fromDataServer = (UniversalMessage) in.readObject()) != null)  {
			    	//get result from data server
			    	//update result in the server
//			    	worker_results[thread_id].add(fromDataServer.getResult());
			    	System.out.println("#result of range query: size = "+ fromDataServer.getResult().size());
			    	//return result to client
                }
	    	    out.close();
	    	    in.close();
//	    	    stdIn.close();
	    	    socket.close();
	        } catch (UnknownHostException e) {
	            System.err.println("Don't know about host: "+host);
	            System.exit(1);
	        } catch (IOException e) {
	            System.err.println("Couldn't get I/O for the connection to: " + host);
	            System.exit(1);
	        } catch (ClassNotFoundException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
	        
            
		}
	}
}
